package com.innodealing.factory;

import com.innodealing.constants.KafkaConstants;
import com.innodealing.model.UserKafkaMessage;
import com.innodealing.serialization.UserKafkaMessageDeserializationSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 * Kafka源工厂类
 * 
 * @author 系统生成
 * @date 2024/12/18
 */
public final class KafkaSourceFactory {
    
    private KafkaSourceFactory() {
        // 私有构造函数，防止实例化
    }
    


    /**
     * 创建用户事件Kafka源
     * 
     * @param kafkaBootstrapServers Kafka服务器地址
     * @return 用户事件Kafka源
     */
    public static KafkaSource<UserKafkaMessage> createUserEventsSource(String kafkaBootstrapServers) {
        return KafkaSource.<UserKafkaMessage>builder()
                .setBootstrapServers(kafkaBootstrapServers)
                .setTopics(KafkaConstants.Topics.USER_EVENTS)
                .setGroupId(KafkaConstants.ConsumerGroups.USER_EVENTS_GROUP)
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setValueOnlyDeserializer(new UserKafkaMessageDeserializationSchema())
                .build();
    }
} 